package com.fudan.run.action;

import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.http.HttpHost;
import org.apache.spark.api.java.JavaRDD;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.script.Script;

import com.fudan.cfg.RunConfiguration;
import com.fudan.cfg.action.SaveToEs;
import com.fudan.cfg.action.UpsertToEs;
import com.fudan.run.VariableRddMap;
import com.fudan.run.ctx.annotation.ActionAdapter;

@ActionAdapter("upsert-to-es")
public class UpsertToEsAction extends ActionHandler<UpsertToEs>{

	@Override
	public void action(UpsertToEs actionNode, VariableRddMap variableRddMap,RunConfiguration runConfiguration) {
		JavaRDD<Map<String, Object>> targetRdd = variableRddMap.getRdd(actionNode.getTarget());
		final Map<String, Object> metadata = ActionMeta.buildMetadata(runConfiguration.getRunId());
		targetRdd.foreachPartition(it->{
			try(RestHighLevelClient client = new RestHighLevelClient(
					RestClient.builder(new HttpHost(actionNode.getHost(), Integer.parseInt(actionNode.getPort()), "http")));) {
				while(it.hasNext()) {
					Map<String, Object> item = it.next();
					item.putAll(metadata);
					StringBuilder code = new StringBuilder();
					String codeTemp = "def node${1} = ctx._source['${2}'];if(node${1} == null) {ctx._source['${2}'] = params.${2};} else{node${1}.addAll(params.${2});}";
					Set<String> keySet = item.keySet();
					int i = 0;
					for (String key : keySet) {
						if(item.get(key) instanceof List) {
							code.append(codeTemp.replaceAll("\\$\\{1\\}", String.valueOf(i++)).replaceAll("\\$\\{2\\}", key));
						}
					}
					UpdateRequest updateRequest = new UpdateRequest(actionNode.getIndex(),String.valueOf(item.get(actionNode.getId())))
					.script(new Script(Script.DEFAULT_SCRIPT_TYPE, Script.DEFAULT_SCRIPT_LANG, code.toString(), item))
					.upsert(item);
					client.update(updateRequest,RequestOptions.DEFAULT);
				}
			}

		});
	}

}
